package com.java.springboot.elasticsearch.controller;

import com.alibaba.fastjson.JSON;
import com.java.springboot.elasticsearch.config.ElasticSearchConfig;
import com.java.springboot.elasticsearch.entity.User;
import lombok.extern.slf4j.Slf4j;
import org.elasticsearch.action.ActionListener;
import org.elasticsearch.action.DocWriteResponse;
import org.elasticsearch.action.bulk.*;
import org.elasticsearch.action.delete.DeleteRequest;
import org.elasticsearch.action.delete.DeleteResponse;
import org.elasticsearch.action.get.GetRequest;
import org.elasticsearch.action.get.GetResponse;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.action.index.IndexResponse;
import org.elasticsearch.action.support.replication.ReplicationResponse;
import org.elasticsearch.action.update.UpdateRequest;
import org.elasticsearch.action.update.UpdateResponse;
import org.elasticsearch.client.RequestOptions;
import org.elasticsearch.client.RestHighLevelClient;
import org.elasticsearch.client.core.GetSourceRequest;
import org.elasticsearch.client.core.GetSourceResponse;
import org.elasticsearch.common.Strings;
import org.elasticsearch.common.unit.ByteSizeUnit;
import org.elasticsearch.common.unit.ByteSizeValue;
import org.elasticsearch.common.xcontent.XContentType;
import org.elasticsearch.core.TimeValue;
import org.elasticsearch.index.get.GetResult;
import org.elasticsearch.search.fetch.subphase.FetchSourceContext;
import org.springframework.web.bind.annotation.*;

import javax.annotation.Resource;
import java.io.IOException;
import java.util.Map;

/**
 * document相关api
 *
 * @author zhangtengfei
 * @date 2021/9/29 15:34
 */
@Slf4j
@RestController
@RequestMapping("/es")
public class EsDocumentController {

    @Resource
    private RestHighLevelClient client;

    /**
     * 插入一条数据
     *
     * @throws IOException
     */
    @PostMapping
    public IndexResponse add() throws IOException {
        /**
         * 1. 指定Index
         */
        IndexRequest request = new IndexRequest("users");
        /**
         * 2. 指定document
         */
        request.id("1");
        /**
         * 3.1 document source
         */
//        request.source("username", "zhangsan", "age", 18, "sex", "male");
        /**
         * 3.2 Map
         */
//        Map<String, Object> map = new HashMap<>();
//        map.put("user", "kimchy");
//        map.put("postDate", new Date());
//        map.put("message", "trying out Elasticsearch");
//        // Document source provided as a Map which gets automatically converted to JSON format
//        request.source(map);
        /**
         * 3.3 json类型
         */
        User user = User.builder().username("zzz").age(18).sex("male").build();
        String jsonStr = JSON.toJSONString(user);
        request.source(jsonStr, XContentType.JSON);

        /**
         * 4.1 同步方式执行
         */
        IndexResponse indexResponse = client.index(request, ElasticSearchConfig.COMMON_OPTIONS);
        /**
         * 4.2 异步方式执行
         */
//        restHighLevelClient.indexAsync(request, ElasticSearchConfig.COMMON_OPTIONS,
//                new ActionListener<IndexResponse>() {
//                    @Override
//                    public void onResponse(IndexResponse indexResponse) {
//                        // 成功执行后 执行代码
//                    }
//
//                    @Override
//                    public void onFailure(Exception e) {
//                        // 失败之后 执行代码
//                    }
//                });
        System.out.println(indexResponse);
        /**
         * 5. IndexResponse 
         */
        String index = indexResponse.getIndex();
        String id = indexResponse.getId();
        if (indexResponse.getResult() == DocWriteResponse.Result.CREATED) {
            // Handle (if needed) the case where the document was created for the first time
        } else if (indexResponse.getResult() == DocWriteResponse.Result.UPDATED) {
            // Handle (if needed) the case where the document was rewritten as it was already existing
        }
        ReplicationResponse.ShardInfo shardInfo = indexResponse.getShardInfo();
        if (shardInfo.getTotal() != shardInfo.getSuccessful()) {
            // Handle the situation where number of successful shards is less than total shards
        }
        if (shardInfo.getFailed() > 0) {
            for (ReplicationResponse.ShardInfo.Failure failure : shardInfo.getFailures()) {
                String reason = failure.reason();
                // Handle the potential failures
            }
        }
        return indexResponse;
    }

    /**
     * 批量操作命令
     *
     * @throws IOException
     */
    @PostMapping("/bulk")
    public BulkResponse bulkIndex() throws IOException {
        BulkRequest request = new BulkRequest();
        request.add(new IndexRequest("posts").id("1").source(XContentType.JSON, "field", "foo"));
        request.add(new IndexRequest("posts").id("2").source(XContentType.JSON, "field", "bar"));
        request.add(new IndexRequest("posts").id("3").source(XContentType.JSON, "field", "baz"));
        // And different operation types can be added to the same BulkRequest
        request.add(new DeleteRequest("posts", "3"));
        request.add(new UpdateRequest("posts", "2").doc(XContentType.JSON, "other", "test"));
        request.add(new IndexRequest("posts").id("4").source(XContentType.JSON, "field", "baz"));

        /**
         * 同步执行
         */
        BulkResponse bulkResponse = client.bulk(request, RequestOptions.DEFAULT);
        /**
         * 异步执行
         */
//        client.bulkAsync(request, RequestOptions.DEFAULT, new ActionListener<BulkResponse>() {
//            @Override
//            public void onResponse(BulkResponse bulkResponse) {
//
//            }
//
//            @Override
//            public void onFailure(Exception e) {
//
//            }
//        });
        /**
         * BulkResponse
         */
        for (BulkItemResponse bulkItemResponse : bulkResponse) {
            DocWriteResponse itemResponse = bulkItemResponse.getResponse();

            switch (bulkItemResponse.getOpType()) {
                case INDEX:
                case CREATE:
                    IndexResponse indexResponse = (IndexResponse) itemResponse;
                    break;
                case UPDATE:
                    UpdateResponse updateResponse = (UpdateResponse) itemResponse;
                    break;
                case DELETE:
                    DeleteResponse deleteResponse = (DeleteResponse) itemResponse;
            }
        }
        return bulkResponse;
    }

    /**
     * The BulkProcessor simplifies the usage of the Bulk API 
     * @throws IOException
     */
    /**
     * 批量操作命令
     *
     * @throws IOException
     */
    @PostMapping("/bulk/processor")
    public void BulkProcessorIndex() throws IOException {
        BulkProcessor.Listener listener = new BulkProcessor.Listener() {
            @Override
            public void beforeBulk(long executionId, BulkRequest request) {
                int numberOfActions = request.numberOfActions();
                log.debug("Executing bulk [{}] with {} requests", executionId, numberOfActions);
            }

            @Override
            public void afterBulk(long executionId, BulkRequest request, BulkResponse response) {
                if (response.hasFailures()) {
                    log.warn("Bulk [{}] executed with failures", executionId);
                } else {
                    log.debug("Bulk [{}] completed in {} milliseconds", executionId, response.getTook().getMillis());
                }
            }

            @Override
            public void afterBulk(long executionId, BulkRequest request, Throwable failure) {
                log.error("Failed to execute bulk", failure);
            }
        };
        // 需要两个组件：RestHighLevelClient BulkProcessor.Listener
        BulkProcessor bulkProcessor = BulkProcessor.builder((request, bulkListener) ->
                client.bulkAsync(request, RequestOptions.DEFAULT, bulkListener), listener, "bulk-processor-name").build();
        BulkProcessor.Builder builder = BulkProcessor.builder((request, bulkListener) ->
                client.bulkAsync(request, RequestOptions.DEFAULT, bulkListener), listener, "bulk-processor-name");
        builder.setBulkActions(500);
        builder.setBulkSize(new ByteSizeValue(1L, ByteSizeUnit.MB));
        builder.setConcurrentRequests(0);
        builder.setFlushInterval(TimeValue.timeValueSeconds(10L));
        builder.setBackoffPolicy(BackoffPolicy.constantBackoff(TimeValue.timeValueSeconds(1L), 3));

        IndexRequest one = new IndexRequest("posts").id("1")
                .source(XContentType.JSON, "title", "In which order are my Elasticsearch queries executed?");
        IndexRequest two = new IndexRequest("posts").id("2")
                .source(XContentType.JSON, "title", "Current status and upcoming changes in Elasticsearch");
        IndexRequest three = new IndexRequest("posts").id("3")
                .source(XContentType.JSON, "title", "The Future of Federated Search in Elasticsearch");
        bulkProcessor.add(one);
        bulkProcessor.add(two);
        bulkProcessor.add(three);
    }

    @GetMapping
    public GetResponse get() throws IOException {
        /**
         * 1. 指定查询的 index、document id
         */
        GetRequest request = new GetRequest("users", "1");
        /**
         * 2. 配置查询 source 的 includes、excludes（指定 field 查询）
         */
        String[] includes1 = new String[]{"username", "sex"};
        String[] includes2 = Strings.EMPTY_ARRAY;
        String[] excludes1 = new String[]{"age"};
        String[] excludes2 = Strings.EMPTY_ARRAY;
        FetchSourceContext fetchSourceContext = new FetchSourceContext(true, includes2, excludes1);
        request.fetchSourceContext(fetchSourceContext);
        /**
         * 3.1 同步查询
         */
        GetResponse getResponse = client.get(request, ElasticSearchConfig.COMMON_OPTIONS);
        /**
         * 3.2 异步查询
         */
//        restHighLevelClient.getAsync(request, ElasticSearchConfig.COMMON_OPTIONS,
//                new ActionListener<GetResponse>() {
//                    @Override
//                    public void onResponse(GetResponse getResponse) {
//                        // Called when the execution is successfully completed.
//                    }
//                    @Override
//                    public void onFailure(Exception e) {
//                        // Called when the whole GetRequest fails.
//                    }
//                });
        System.out.println(getResponse);
        /**
         * 4. GetResponse
         */
        String index = getResponse.getIndex();
        String id = getResponse.getId();
        if (getResponse.isExists()) {
            long version = getResponse.getVersion();
            // Retrieve the document as a String
            String sourceAsString = getResponse.getSourceAsString();
            // Retrieve the document as a Map<String, Object>
            Map<String, Object> sourceAsMap = getResponse.getSourceAsMap();
            // Retrieve the document as a byte[]
            byte[] sourceAsBytes = getResponse.getSourceAsBytes();
        } else {
            // Handle the scenario where the document was not found
        }

        return getResponse;
    }

    @GetMapping("/source")
    public Map<String, Object> getSource() throws IOException {
        /**
         * 1. 指定查询的 index、document id
         */
        GetSourceRequest request = new GetSourceRequest("users", "1");
        /**
         * 2. 配置查询 source 的 includes、excludes（指定 field 查询）
         */
        String[] includes = Strings.EMPTY_ARRAY;
        String[] excludes = new String[]{"age"};
        FetchSourceContext fetchSourceContext = new FetchSourceContext(true, includes, excludes);
        request.fetchSourceContext(fetchSourceContext);
        /**
         * 3.1 同步查询
         */
        GetSourceResponse response = client.getSource(request, ElasticSearchConfig.COMMON_OPTIONS);
        /**
         * 3.2 异步查询
         */
//        restHighLevelClient.getSourceAsync(request, RequestOptions.DEFAULT,
//                new ActionListener<GetSourceResponse>() {
//                    @Override
//                    public void onResponse(GetSourceResponse getResponse) {
//
//                    }
//
//                    @Override
//                    public void onFailure(Exception e) {
//
//                    }
//                });
        Map<String, Object> source = response.getSource();
        return source;
    }

    /**
     * returns true if a document exists, and false otherwise.
     *
     * @return
     * @throws IOException
     */
    @GetMapping("/exist")
    public boolean exist() throws IOException {
        /**
         * 1. 指定查询的 index、document id
         */
        GetRequest request = new GetRequest("users", "1");
        /**
         * 2. Disable fetching _source
         */
        FetchSourceContext fetchSourceContext = new FetchSourceContext(false);
        request.fetchSourceContext(fetchSourceContext);
        request.storedFields("_none_");
        /**
         * 3.1 同步执行
         */
        boolean exists = client.exists(request, RequestOptions.DEFAULT);
        /**
         * 3.2 异步执行
         */
//        client.existsAsync(request, RequestOptions.DEFAULT,
//                new ActionListener<Boolean>() {
//                    @Override
//                    public void onResponse(Boolean exists) {
//                    }
//                    @Override
//                    public void onFailure(Exception e) {
//                    }
//                });
        return exists;
    }

    @PutMapping
    public UpdateResponse update() throws IOException {
        /**
         * 1. 指定 index、document id
         */
        UpdateRequest request = new UpdateRequest("posts", "1");
        /**
         * 2. updates with a script / as a stored script
         */
//        Map<String, Object> parameters = singletonMap("count", 4);
//        Script inline = new Script(ScriptType.INLINE, "painless",
//                "ctx._source.field += params.count", parameters);
//        request.script(inline);

//        Script stored = new Script(ScriptType.STORED, null, "increment-field", parameters);
//        request.script(stored);
        /**
         * 3. 设置要更新的 document 内容
         */
        request.doc("username", "zhangsan - update");
        /**
         * 4.1 同步执行
         */
        UpdateResponse updateResponse = client.update(request, RequestOptions.DEFAULT);
        /**
         * 4.2 异步执行
         */
//        client.updateAsync(request, RequestOptions.DEFAULT,
//                new ActionListener<UpdateResponse>() {
//                    @Override
//                    public void onResponse(UpdateResponse updateResponse) {
//
//                    }
//
//                    @Override
//                    public void onFailure(Exception e) {
//
//                    }
//                });
        /**
         * 5. UpdateResponse
         */
        String index = updateResponse.getIndex();
        String id = updateResponse.getId();
        long version = updateResponse.getVersion();
        if (updateResponse.getResult() == DocWriteResponse.Result.CREATED) {

        } else if (updateResponse.getResult() == DocWriteResponse.Result.UPDATED) {

        } else if (updateResponse.getResult() == DocWriteResponse.Result.DELETED) {

        } else if (updateResponse.getResult() == DocWriteResponse.Result.NOOP) {

        }
        GetResult result = updateResponse.getGetResult();
        if (result.isExists()) {
            String sourceAsString = result.sourceAsString();
            Map<String, Object> sourceAsMap = result.sourceAsMap();
            byte[] sourceAsBytes = result.source();
        } else {

        }
        /**
         * It is also possible to check for shard failures:
         */
        ReplicationResponse.ShardInfo shardInfo = updateResponse.getShardInfo();
        if (shardInfo.getTotal() != shardInfo.getSuccessful()) {

        }
        if (shardInfo.getFailed() > 0) {
            for (ReplicationResponse.ShardInfo.Failure failure :
                    shardInfo.getFailures()) {
                String reason = failure.reason();
            }
        }
        return updateResponse;
    }

    @DeleteMapping
    public DeleteResponse delete() throws IOException {
        /**
         * 1. 指定 index、document id
         */
        DeleteRequest request = new DeleteRequest("users", "1");
        /**
         * 2.1 同步执行
         */
        DeleteResponse deleteResponse = client.delete(request, ElasticSearchConfig.COMMON_OPTIONS);
        System.out.println(deleteResponse);
        /**
         * 2.2 异步执行
         */
//        client.deleteAsync(request, RequestOptions.DEFAULT,
//                new ActionListener<DeleteResponse>() {
//                    @Override
//                    public void onResponse(DeleteResponse deleteResponse) {
//                    }
//
//                    @Override
//                    public void onFailure(Exception e) {
//                    }
//                });
        /**
         * 3. DeleteResponse
         */
        String index = deleteResponse.getIndex();
        String id = deleteResponse.getId();
        long version = deleteResponse.getVersion();
        ReplicationResponse.ShardInfo shardInfo = deleteResponse.getShardInfo();
        if (shardInfo.getTotal() != shardInfo.getSuccessful()) {

        }
        if (shardInfo.getFailed() > 0) {
            for (ReplicationResponse.ShardInfo.Failure failure :
                    shardInfo.getFailures()) {
                String reason = failure.reason();
            }
        }
        /**
         * check whether the document was found or not
         */
        if (deleteResponse.getResult() == DocWriteResponse.Result.NOT_FOUND) {

        }
        return deleteResponse;
    }
}
